package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo2Parition {
  def main(args: Array[String]): Unit = {

    //创建spark环境
    val conf = new SparkConf()

    conf.setAppName("partition")
    conf.setMaster("local")

    val sc = new SparkContext(conf)

    //读取文件得到一个RDD
    val linesRDD: RDD[String] = sc.textFile("data/words", minPartitions = 4)

    /**
      * 默认一个block对应一个分区， 切片规则和mr是一样的
      * 当设置了minPartitions,rdd的分数数会大于等于这个值，具体多少分区需要通过计算得到
      * 计算原则是保证每一个切片的数据量是差不多的
      */
    println(s"lineRDD:${linesRDD.getNumPartitions}")

    /**
      * 如果算子没有产生shuffle，返回的新的rdd的分区数默认等于前一个rdd的分区数
      *
      */

    val wordsRDD: RDD[String] = linesRDD.flatMap(line => line.split(","))

    println(s"wordsRDD:${wordsRDD.getNumPartitions}")

    /**
      * 如果算子产生了shuffle,可以手动设计分区数据， 默认也是等于前一个rdd的分区数
      * 类似在MR中设置reduce的数量
      *
      * 一般来说保证每一个分区类的数据量在128m左右是比较合适的
      * 如果资源无限的情况下，分区越多，执行的并发越高，效率越高
      *
      */

    val groupByRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy((word: String) => word, numPartitions = 2)

    println(s"groupByRDD:${groupByRDD.getNumPartitions}")

    groupByRDD.foreach(println)


    while (true) {

    }

  }

}
